// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Author: wsfuyibing <682805@qq.com>
// Date: 2024-12-09

package managers

import (
	"context"
	"gitee.com/go-libs/log"
	"gitee.com/gomq/sr"
	"gomq/app"
	"gomq/app/errs"
	"gomq/app/models"
	"gomq/app/services"
	"net/http"
	"sync"
	"time"
)

var consumingPool = &sync.Pool{
	New: func() any {
		return &consuming{}
	},
}

type (
	// Consuming is an interface used to consume received messages from
	// queue like rabbitmq, rocketmq and so on.
	Consuming interface {
		// Do process received message.
		Do() (err error)

		// Release put consuming instance to pool.
		Release()
	}

	consuming struct {
		isNotification bool
		m              Manager
		msg            *sr.Message
		sm             ServerManager
		sub            sr.Subscription
	}
)

// NewConsuming acquires a consuming instance from pool.
func NewConsuming(m Manager, sm ServerManager, sub sr.Subscription, msg *sr.Message) Consuming {
	return consumingPool.
		Get().(*consuming).
		acquirer(m, sm, sub, msg)
}

func (o *consuming) Do() (err error) {
	var (
		body         []byte
		ctx          = o.msg.Ctx
		endpoint     sr.Endpoint
		has          bool
		filtered     bool
		notification *sr.Notification
		response     sr.Response
		scheduler    sr.Scheduler
	)

	log.Infofc(ctx, `[consuming-manager] invoked: called by %s`, o.sm.Cluster().Plugin)

	// Invoked before executor done.
	defer func() {
		if r := recover(); r != nil {
			log.Fatalfc(ctx, `[consuming-manager] consuming caused an panic: %v`, r)
		}

		// Invoke: call notifier to send notification message.
		if response != nil && !o.isNotification {
			if response.HasError() {
				if o.msg.ConsumedCount == o.sub.GetMaxRetry() {
					if _, ok := o.sub.GetFailedEndpoint(); ok {
						o.DoNotifier(ctx, response)
					}
				}
			} else {
				if _, ok := o.sub.GetSucceedEndpoint(); ok {
					o.DoNotifier(ctx, response)
				}
			}
		}

		// Invoke: store into message and snapshot tables if enabled.
		if o.sub.GetStore() {
			cts := log.NewContextForChildInfo(ctx, `[consuming-manager] storage invoked`)
			if response != nil {
				o.DoStoreMessage(cts, filtered, response.HasError())
				o.DoStoreSnapshot(cts, endpoint, response)
			} else {
				o.DoStoreMessage(cts, filtered, false)
			}
		}

		// Invoke: release instances.
		if response != nil {
			response.Release()
		}
	}()

	// Invoke: pre check on.
	if o.isNotification {
		// Filtered: by notification verify.
		if endpoint, notification, err = o.DoNotification(); err != nil {
			log.Warnfc(ctx, `[consuming-manager] discard notification message: for="%v"`, err)
			err = nil
			filtered = true
			return
		}

		// Reset scheduler body.
		body = notification.Byte()
	} else {
		// Filtered: endpoint not specified.
		if endpoint, has = o.sub.GetEndpoint(); !has {
			filtered = true
			log.Warnfc(ctx, `[consuming-manager] discard message: for="endpoint not configured in subscription task"`)
			return
		}

		// Filtered: message body not matched on filter conditions.
		if filtered = o.DoFilter(ctx); filtered {
			log.Infofc(ctx, `[consuming-manager] discard message for="filter conditions not matched"`)
			return
		}

		// Reset scheduler body.
		body = o.msg.Body
	}

	// Invoke: create scheduler
	if scheduler, err = o.m.Supporter().NewScheduler(endpoint.GetProtocol()); err != nil {
		log.Warnfc(ctx, `[consuming-manager] discard message: for="%v"`, err)
		err = nil
		filtered = true
		return
	}

	// Invoke: call scheduler.
	response, err = o.DoSchedule(ctx, scheduler, endpoint, body)
	return
}

func (o *consuming) DoFilter(ctx context.Context) (filtered bool) {
	var (
		filter sr.Filter
		ok     bool
	)

	// Do nothing if filter not configured by task.condition_format and
	// task.condition_expression columns on table.
	if filter, ok = o.sub.GetFilter(); !ok {
		return
	}

	// Ignore filter by error condition expression or format.
	if filter.HasError() {
		log.Warnfc(ctx, `[consuming-manager] filter invoked: error="%v"`, filter.GetError())
		return
	}

	// Check filter with message body.
	ctx = log.NewContextForChildInfo(ctx, `[consuming-manager] filter invoked: for="message body", expression="%s"`, filter.Query())
	if !filter.Match(ctx, o.msg.Body) {
		filtered = true
	}
	return
}

func (o *consuming) DoNotification() (endpoint sr.Endpoint, notification *sr.Notification, err error) {
	var (
		has bool
		sub sr.Subscription
	)

	notification = sr.NewNotification()

	// Return err if parse message body as notification.
	if err = notification.Unmarshal(o.msg.Body); err != nil {
		return
	}

	// Return err if origin subscription task deleted or disabled.
	if sub, has = o.m.MemoryManager().GetSubscription(notification.SubscriptionId); !has {
		err = errs.ErrSubscriptionNoInMemory
		return
	}

	// Failed notification.
	if o.msg.Tag.IsFailed() {
		if endpoint, has = sub.GetFailedEndpoint(); !has {
			err = errs.ErrEndpointForFailedNotFound
		}
		return
	}

	// Succeed notification.
	if o.msg.Tag.IsSucceed() {
		if endpoint, has = sub.GetSucceedEndpoint(); !has {
			err = errs.ErrEndpointForSucceedNotFound
		}
		return
	}

	// Unknown
	// notification message.
	err = errs.ErrUnsupportedNotificationMessage
	return
}

func (o *consuming) DoNotifier(ctx context.Context, response sr.Response) {
	var (
		ntf *sr.Notification
		pay *sr.Payload
	)

	log.Infofc(ctx, `[consuming-manager] notifier invoked: publish scheduling result notification`)

	// Acquires a notification message instance and release to pool when
	// publisher done.
	ntf = sr.NewNotification()
	ntf.SubscriptionId = o.sub.GetId()
	ntf.ResponseBody = response.GetBody()
	defer ntf.Release()

	// Acquires a payload instance from pool.
	pay = sr.NewPayloadWithHeader(o.msg.Header)
	pay.Body = ntf.Byte()
	pay.Topic = sr.ReadNotificationTopic()

	// Set message tag for notification.
	if response.HasError() {
		pay.Tag = sr.ReadFailedNotificationTag()
	} else {
		pay.Tag = sr.ReadSucceedNotificationTag()
	}

	// Call publisher to publish.
	if err := o.sm.Publish(pay); err != nil {
		_, err = o.sm.Servlet().Producer().Publish(ctx, pay)
	}
}

func (o *consuming) DoStoreMessage(ctx context.Context, filtered, failed bool) {
	var (
		affects   int64
		bean, req *models.Messages
		err       error
		service   = services.NewMessagesService()
	)

	// Read previous message from database.
	if bean, err = service.GetBySubscriptionIdAndMessageId(ctx, o.sub.GetId(), o.msg.MessageId); err != nil {
		return
	}

	// Update status if exists.
	if bean != nil {
		if filtered {
			affects, err = service.SetStatusAsFiltered(ctx, bean.Id)
		} else {
			if failed {
				affects, err = service.SetStatusAsFailed(ctx, bean.Id)
			} else {
				affects, err = service.SetStatusAsSucceed(ctx, bean.Id)
			}
		}
		if err == nil {
			log.Infofc(ctx, `[consuming-manager] messages save succeed: primary-key="%d", affects="%d"`, bean.Id, affects)
		}
		return
	}

	// Set messages request params.
	req = &models.Messages{
		Topic:          o.sub.GetTopic(),
		Tag:            o.sub.GetTag(),
		SubscriptionId: o.sub.GetId(),
		MessageId:      o.msg.MessageId,
		MessageTime:    o.msg.Time.Format(time.DateTime),
		MessageBody:    o.msg.Body,
	}

	// Load trace id from context.
	if tracing := log.ContextTracing(ctx); tracing != nil {
		req.TraceId = tracing.GetTraceId().String()
	}

	// Set messages status.
	if filtered {
		req.Status = app.MessageStatusFiltered
	} else {
		if failed {
			req.Status = app.MessageStatusFailed
		} else {
			req.Status = app.MessageStatusSucceed
		}
	}

	// Add messages into database.
	if bean, err = service.Add(ctx, req); err == nil {
		log.Infofc(ctx, `[consuming-manager] messages save succeed: primary-key="%d"`, bean.Id)
	}
}

func (o *consuming) DoStoreSnapshot(ctx context.Context, endpoint sr.Endpoint, response sr.Response) {
	var (
		bean, req *models.Snapshots
		err       error
	)

	// Set snapshot request params.
	req = &models.Snapshots{
		Duration:       response.GetDuration().Milliseconds(),
		Topic:          o.sub.GetTopic(),
		Tag:            o.sub.GetTag(),
		Handler:        endpoint.GetAddress(),
		SubscriptionId: o.sub.GetId(),
		ReceivedTime:   o.msg.ConsumedTime.Format(time.DateTime),
		MessageId:      o.msg.MessageId,
		MessageTime:    o.msg.Time.Format(time.DateTime),
		MessageBody:    o.msg.Body,
		ResponseBody:   response.GetBody(),
	}

	// Set
	// snapshot status.
	if response.HasError() {
		// Failed.
		req.ResponseError = []byte(response.GetError().Error())
		req.Status = app.MessageStatusFailed
	} else {
		// Succeed.
		req.Status = app.MessageStatusSucceed
	}

	// Add snapshot into table.
	if bean, err = services.NewSnapshotsService().Add(ctx, req); err == nil {
		log.Infofc(ctx, `[consuming-manager] snapshots save succeed: primary-key="%d"`, bean.Id)
	}
}

func (o *consuming) DoSchedule(ctx context.Context, scheduler sr.Scheduler, endpoint sr.Endpoint, body []byte) (response sr.Response, err error) {
	ctx = log.NewContextForChildInfo(ctx, `[consuming-manager] scheduler invoked: protocol="%s"`, endpoint.GetProtocol())

	// Invoked when scheduler done.
	defer func() {
		// Catch runtime panic.
		if r := recover(); r != nil {
			log.Panicfc(ctx, `[consuming-manager] scheduling caused an panic: %v`)

			// Restore error reason.
			if err == nil {
				err = sr.NewError(http.StatusInternalServerError, `%v`, r)
				response.SetError(err)
			}
		}
	}()

	// Invoke: call dispatcher of the scheduler to dispatch message.
	if response = scheduler.Dispatch(ctx, endpoint, o.msg, body); response.HasError() {
		err = response.GetError()
		return
	}

	// Invoke: filter response body if enabled.
	if filter, ok := endpoint.GetFilter(); ok {
		cts := log.NewContextForChildInfo(ctx, `[consuming-manager] filter invoked: for="response body", expression="%s"`, filter.Query())

		// Restore error reason with filter.
		if filter.HasError() {
			err = filter.GetError()
			response.SetError(err)
			return
		}

		// Restore error reason if response body not matched on filter
		// conditions.
		if !filter.Match(cts, response.GetBody()) {
			err = errs.ErrFilterResponseBodyNotMatched
			response.SetError(err)
		}
	}
	return
}

func (o *consuming) Release() { o.releaser() }

// + Access

func (o *consuming) acquirer(m Manager, sm ServerManager, sub sr.Subscription, msg *sr.Message) *consuming {
	o.m, o.sm, o.sub, o.msg = m, sm, sub, msg
	o.isNotification = msg.Topic.IsNotification()
	return o
}

func (o *consuming) releaser() {
	o.msg, o.m, o.sm, o.sub = nil, nil, nil, nil
	consumingPool.Put(o)
}
